package org.quxiao.mmrsql.executor.reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TableJoinReducer extends Reducer<Text, Text, Text, Text> {

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		/*
		 * attr1: t1:id1, t1:id2, t2:id1, t2:id3, ...
		 *
		 * 
		 */
//		System.out.println("key: " + key.toString());
//		System.out.println("======================\nvalues:");
//		for (Text t: values)
//		{
//			System.out.print(t.toString() + ", ");
//		}
//		System.out.println("\n======================\n");
//		
		List<String> list1 = new ArrayList<String>();
		List<String> list2 = new ArrayList<String>();
		String table1 = context.getConfiguration().get("table1");
//		String table2 = context.getConfiguration().get("table2");
		for (Text v: values)
		{
			String str = v.toString();
			String table = str.split(":", 2)[0];
			if ( table.compareTo(table1) == 0 )
				list1.add(str);
			else
				list2.add(str);
		}
		for (String str1: list1)
		{
			for (String str2: list2)
			{
				context.write(new Text(str1 + "," + str2), new Text(key));
			}
		}
		
	}

}
